package flink.demo;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.util.List;

public class TransformationDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
//        environment.setRuntimeMode(RuntimeExecutionMode.BATCH);
        DataStreamSource<String> streamSource = environment.readTextFile("data/transformation_input/userinfo.txt");
        SingleOutputStreamOperator<UserInfo> mapStream = streamSource.map(json -> JSON.parseObject(json, UserInfo.class));



        /**
         * filter算子的演示
         *   请过滤掉好友超过3位的用户数据
         */

       // SingleOutputStreamOperator<UserInfo> filter = map.filter(userInfo -> userInfo.getFriends().size() <= 3);


        /**
         * flatmap算子的演示
         *  把每个用户的好友信息，全部提取出来（带上用户自身的信息），并且压平，放入结果流中
         *  {"uid":1,"name":"zs","gender":"male","friends":[{"fid":2,"name":"aa"},{"fid":3,"name":"bb"}]}
         *  =>
         *  {"uid":1,"name":"zs","gender":"male","fid":2,"fname":"aa"}
         *  {"uid":1,"name":"zs","gender":"male","fid":3,"fname":"bb"}
         */

        SingleOutputStreamOperator<UserFriendInfo> flatMap = mapStream.flatMap(new FlatMapFunction<UserInfo, UserFriendInfo>() {
            @Override
            public void flatMap(UserInfo userInfo, Collector<UserFriendInfo> collector) throws Exception {
                List<FriendInfo> friends = userInfo.getFriends();
                friends.forEach(friendInfo -> {
                    collector.collect(new UserFriendInfo(userInfo.getUid(), userInfo.getName(), userInfo.getGender(), friendInfo.getFid(), friendInfo.getName()));
                });
            }
        });
        /**
         * keyBy算子的演示
         * 对上一步的结果，按用户性别分组
         *
         * 滚动聚合算子（只能在 KeyedStream 流上调用）：  sum算子 、 min算子 、 minBy算子 、 max算子、  maxBy算子、 reduce算子的演示
         * 并统计：
         *    各性别用户的好友总数
         *
         *    各性别中，用户好友数量最大值
         *    各性别中，用户好友数量最小值
         *
         *    求各性别中，好友数量最大的那个人
         *    求各性别中，好友数量最小的那个人
         *
         */

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = flatMap.map(new MapFunction<UserFriendInfo, Tuple2<String, Integer>>() {


            @Override
            public Tuple2<String, Integer> map(UserFriendInfo userFriendInfo) throws Exception {
                return Tuple2.of(userFriendInfo.getGender(), 1);
            }
        })

                .keyBy(new KeySelector<Tuple2<String, Integer>, String>() {

                    @Override
                    public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                        return stringIntegerTuple2.f0;
                    }
                });


        /**
         * max / maxBy  都是滚动聚合：  算子不会把收到的所有数据全部攒起来；而是只在状态中记录上一次的聚合值，然后当新数据到达的时候，会根据逻辑去更新 状态中记录的聚合值，并输出最新状态数据
         * max / maxBy  区别： 更新状态的逻辑！  max只更新要求最大值的字段；  而 maxBy 会更新整条数据；
         * 同理，min和minBy也如此
         */


        SingleOutputStreamOperator<Tuple4<Integer, String, String, Integer>> streamOperator = mapStream.map(new MapFunction<UserInfo, Tuple4<Integer, String, String, Integer>>() {

            @Override
            public Tuple4<Integer, String, String, Integer> map(UserInfo userInfo) throws Exception {
                return Tuple4.of(userInfo.getUid(), userInfo.getName(), userInfo.getGender(), userInfo.getFriends().size());
            }
        }).keyBy(new KeySelector<Tuple4<Integer, String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple4<Integer, String, String, Integer> value) throws Exception {
                return value.f2;
            }
        }).maxBy(3);
//                .keyBy(tp -> tp.f2)
//                /*.max(3); */              // 各性别中，用户好友数量最大值
//                .maxBy(3);  // 求各性别中，好友数量最大的那个人





        /**
         * reduce 算子 使用演示
         * 需求： 求各性别中，好友数量最大的那个人，而且如果前后两个人的好友数量相同，则输出的结果中，也需要将uid/name等信息更新成后面一条数据的值
         *
         */

        SingleOutputStreamOperator<Tuple4<Integer, String, String, Integer>> singleOutputStreamOperator = streamOperator.keyBy(new KeySelector<Tuple4<Integer, String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple4<Integer, String, String, Integer> val) throws Exception {
                return val.f2;
            }
        }).reduce(new ReduceFunction<Tuple4<Integer, String, String, Integer>>() {
            /**
             *
             * @param value1  是此前的聚合结果
             * @param value2  是本次的新数据
             * @return 更新后的聚合结果
             * @throws Exception
             */
            @Override
            public Tuple4<Integer, String, String, Integer> reduce(Tuple4<Integer, String, String, Integer> value1, Tuple4<Integer, String, String, Integer> value2) throws Exception {
                if (value1 == null || value2.f3 > value1.f3) {
                    return value2;
                } else {

                    return value1;
                }

            }
        });

        /**
         * 用reduce，来实现sum算子的功能
         * 求：  对上面的  4元组数据  1,ua,male,2  ，求各性别的好友数总和
         * TODO
         */
        SingleOutputStreamOperator<Tuple4<Integer, String, String, Integer>> tuple4SingleOutputStreamOperator = streamOperator.keyBy(new KeySelector<Tuple4<Integer, String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple4<Integer, String, String, Integer> val) throws Exception {
                return val.f2;
            }
        }).reduce(new ReduceFunction<Tuple4<Integer, String, String, Integer>>() {
            @Override
            public Tuple4<Integer, String, String, Integer> reduce(Tuple4<Integer, String, String, Integer> t, Tuple4<Integer, String, String, Integer> t1) throws Exception {
                t1.setField(t1.f3 + (t == null ? 0 : t.f3), 3);
                return t1;
            }
        });
        SingleOutputStreamOperator<Tuple4<Integer, String, String, Integer>> reduce = tuple4SingleOutputStreamOperator;


        environment.execute();
    }
}

class UserInfo implements Serializable {
    private int uid;
    private String name;
    private String gender;
    private List<FriendInfo> friends;

    public int getUid() {
        return uid;
    }

    public void setUid(int uid) {
        this.uid = uid;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getGender() {
        return gender;
    }

    public void setGender(String gender) {
        this.gender = gender;
    }

    public List<FriendInfo> getFriends() {
        return friends;
    }

    public void setFriends(List<FriendInfo> friends) {
        this.friends = friends;
    }

    @Override
    public String toString() {
        return "UserInfo{" +
                "uid=" + uid +
                ", name='" + name + '\'' +
                ", gender='" + gender + '\'' +
                ", friends=" + friends +
                '}';
    }
}


class FriendInfo implements Serializable {
    private int fid;
    private String name;

    public int getFid() {
        return fid;
    }

    @Override
    public String toString() {
        return "FriendInfo{" +
                "fid=" + fid +
                ", name='" + name + '\'' +
                '}';
    }

    public void setFid(int fid) {
        this.fid = fid;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }
}


class UserFriendInfo implements Serializable {

    public UserFriendInfo(int uid, String name, String gender, int fid, String fname) {
        this.uid = uid;
        this.name = name;
        this.gender = gender;
        this.fid = fid;
        this.fname = fname;
    }

    @Override
    public String toString() {
        return "UserFriendInfo{" +
                "uid=" + uid +
                ", name='" + name + '\'' +
                ", gender='" + gender + '\'' +
                ", fid=" + fid +
                ", fname='" + fname + '\'' +
                '}';
    }

    private int uid;
    private String name;
    private String gender;
    private int fid;
    private String fname;

    public int getUid() {
        return uid;
    }

    public void setUid(int uid) {
        this.uid = uid;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getGender() {
        return gender;
    }

    public void setGender(String gender) {
        this.gender = gender;
    }

    public int getFid() {
        return fid;
    }

    public void setFid(int fid) {
        this.fid = fid;
    }

    public String getFname() {
        return fname;
    }

    public void setFname(String fname) {
        this.fname = fname;
    }

}